Today we're gonna be building an Options trading bot using Interactive Brokers and of course Python.

First let's explain options, they are a derivative, not an asset class rather than traditional stocks where they're basically contracts that give you the right but not the obligation to purchase a stock at a future price. Basically, they're a leveraged way of buying different assets and they're incredibly risky. They're much more riskier than stocks because they do have an expiry date as well as leverage applied to them because options pricing varies very differently than a stock price. Okay. So this trading system is obviously gonna be a lot riskier than what we usually do, but I'm gonna be showing you how to use the Interactive Brokers API with python to build an options trading system. So let's hop in and get started.

Prefer video ? Watch on Youtube here : https://youtu.be/yX9zmYdqBEI

Okay, so let's start building our options trading bot in python and with Interactive Brokers. So I've created a an empty python file here called risky options bot. And at the end of this guide there will be a link to the get up if you just want to clone the repository and use it for yourself as well as instructions.

First we start off with our imports, these are libraries or external code needed. We need datetime, ib_insync, apscheduler and asyncio.

#Imports

from datetime import datetime

from ib_insync import *

from apscheduler.schedulers.background import BackgroundScheduler

import asyncio

Now we're gonna connect to Interactive Brokers so we can place trades, be sure to encapsulate this in a Python try, except to handle most errors.

class RiskyOptionsBot:

  """

  Risky Options Bot (Python, Interactive Brokers)

  Buy 2 DTE SPY Contracts on 3 consecutive 5-min higher closes and profit target on next bar

  """

  #Initialize variables

  def __init__(self):

    print("Options Bot Running, connecting to IB ...")

    #Connect to IB

    try:

      self.ib = IB()

      self.ib.connect('127.0.0.1',7496,clientId=1)

      print("Successfully connected to IB")

    except Exception as e:

      print(str(e))

Now let's get our market data creates by contract. So we're gonna make a new class level variable called underlying and for now set it to SPY, which is a S&P500 ETF. By setting keepUpToDate we can get streaming data after we backfill to catchup with market data.

# Create SPY Contract

    self.underlying = Stock('SPY', 'SMART', 'USD')

    self.ib.qualifyContracts(self.underlying)

    print("Backfilling data to catchup ...")

    # Request Streaming bars

    self.data = self.ib.reqHistoricalData(self.underlying,

      endDateTime='',

      durationStr='2 D',

      barSizeSetting='1 min',

      whatToShow='TRADES',

      useRTH=False,

      keepUpToDate=True,)

Let's set a Python class level variable for keep track of our trades.

#Local vars

    self.in_trade = False

Now let's get our options chains and update the chain data every hour. We will also set our on bar update to control logic every time we get need candle stick data.

#Get current options chains

    self.chains = self.ib.reqSecDefOptParams(self.underlying.symbol, '', self.underlying.secType, self.underlying.conId)

    #Update Chains every hour - can't update chains in event loop causes asyncio issues

    update_chain_scheduler = BackgroundScheduler(job_defaults={'max_instances': 2})

    update_chain_scheduler.add_job(func=self.update_options_chains,trigger='cron', hour='*')

    update_chain_scheduler.start()

    print("Running Live")

    # Set callback function for streaming bars

    self.data.updateEvent += self.on_bar_update

    self.ib.execDetailsEvent += self.exec_status

    #Run forever

    self.ib.run()

Let's define our options chain update function, so we can update strike prices and whatnot.

  #Update options chains

  def update_options_chains(self):

    try:

      loop = asyncio.new_event_loop()

      asyncio.set_event_loop(loop)

      print("Updating options chains")

       #Get current options chains

      self.chains = self.ib.reqSecDefOptParams(self.underlying.symbol, '', self.underlying.secType, self.underlying.conId)

      print(self.chains)

    except Exception as e:

      print(str(e))

Now the majority of the logic in on bar update, we are going to buy a SPY call when we see 3 consecutive higher closes and sell as soon as we are in profit.

#On Bar Update, when we get new data

  def on_bar_update(self, bars: BarDataList, has_new_bar: bool):

    try:

      if has_new_bar:

        #Convert BarDataList to pandas Dataframe

        df = util.df(bars)

        # Check if we are in a trade

        if not self.in_trade:

          print("Last Close : " + str(df.close.iloc[-1]))

          #Check for 3 Consecutive Highs

          if df.close.iloc[-1] > df.close.iloc[-2] and df.close.iloc[-2] > df.close.iloc[-3]:

            #Found 3 consecutive higher closes get call contract that's $5 higher than underlying

            for optionschain in self.chains:

              for strike in optionschain.strikes:

                if strike > df.close.iloc[-1] + 5 : #Make sure the strike is $5 away so it's cheaper

                  print("Found 3 consecutive higher closers, entering trade.")

                  self.options_contract = Option(self.underlying.symbol, optionschain.expirations[1], strike, 'C', 'SMART', tradingClass=self.underlying.symbol)

                  # We are not in a trade - Let's enter a trade

                  options_order = MarketOrder('BUY', 1,account=self.ib.wrapper.accounts[-1])

                  trade = self.ib.placeOrder(self.options_contract, options_order)

                  self.lastEstimatedFillPrice = df.close.iloc[-1]

                  self.in_trade = not self.in_trade

                  return # important so it doesn't keep looping

        else: #We are in a trade

          if df.close.iloc[-1] > self.lastEstimatedFillPrice:

            #Sell for profit scalping

            print("Scalping profit.")

            options_order = MarketOrder('SELL', 1,account=self.ib.wrapper.accounts[-1])

            trade = self.ib.placeOrder(self.options_contract, options_order)

    except Exception as e:

      print(str(e))

Lastly let's have a confirmation if an order has been filled for debugging purposes and instantiate our class to actually run the thing!

  #Order Status

  def exec_status(self,trade: Trade,fill: Fill):

    print("Filled")

#Instantiate Class to get things rolling

RiskyOptionsBot()

Full Code : https://github.com/Jake0303/RiskyOptionsBot

My Trading Brokerage : https://www.interactivebrokers.com/mkt/?src=jacobamaraly2&url=%2Fen%2Findex.php%3Ff%3D1338

Discord : https://discord.gg/X6anb8ycKD

Get 3 Trading Algos for free 👨‍💻: https://codingtips.jacobamaral.com/join

Generate automated trading strategies with no coding experience : https://stratgen.io/join

https://youtu.be/yX9zmYdqBEI